package bsp.scan;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.LongPair;
import util.ScanList;
import util.ScanMessage;
import util.Tool;
import util.functors.OpScan;
import bsp.CMapper;

public class ScanStepTwo<T extends Number> extends
		CMapper<LongWritable, Text, Text, NullWritable, ScanMessage> {
	private long chunkey;
//	private String myName;
//	private int myId;
//	private int numPeers;

	private OpScan<T> plus;
	
	public ScanStepTwo(OpScan<T> op){
		plus = op;
	}
	
	protected double str2Num(String token){
		return  Double.parseDouble(token);
	}
	protected String Num2Str(T val){
		return val.toString();
	}
	
	public void map(
			BSPPeer<LongWritable, Text, Text, NullWritable, ScanMessage> contex)
			throws IOException, SyncException, InterruptedException {

		LongWritable key = new LongWritable(-1);
		T sum = plus.id();
		Text line = new Text();

		while (contex.readNext(key, line)) {
			if (chunkey < 0) {
				chunkey = key.get();
				ScanMessage received;
				ScanList<Double> psums = new ScanList<Double>();
				while ((received = contex.getCurrentMessage()) != null) {
					psums.put(received.getMsgId(),  Double.parseDouble(received.getMsg()) );
				}
				Tool.scan(psums,plus);
				sum = plus.id();
				if (psums.floorEntry(chunkey) != null) {
					if (psums.floorEntry(chunkey).getKey() < chunkey)
						sum = (T) psums.floorEntry(chunkey).getValue();
				}

			}
			StringTokenizer itr = new StringTokenizer(line.toString().trim());
			StringBuffer outLine = new StringBuffer(1024);
			while (itr.hasMoreTokens()) {
				//sum += Integer.parseInt(itr.nextToken());
				sum = plus.evaluate(sum, str2Num(itr.nextToken()));
				outLine.append(sum + " ");
			}
			contex.write(new Text(outLine.toString().trim()),
					NullWritable.get());
		}

	}

}